package com.zysd.collection.scheduler.system.job;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.lensyn.collect.common.enums.CollectInfos;
import com.lensyn.collect.common.enums.TaskType;
import com.lensyn.collect.common.types.interfaces.Pipeline;
import com.lensyn.collect.meters.msgs.DataTypeFactory;
import com.lensyn.collect.meters.msgs.TdCurve;
import com.lensyn.collect.meters.msgs.TdDay;
import com.lensyn.collect.meters.msgs.TdMonth;
import com.lensyn.collect.meters.store.Message;
import com.lensyn.collect.meters.store.MetersMultiMessages;
import com.lensyn.collect.meters.type.FreezeThickness;
import com.lensyn.common.utils.domain.common.PageResponse;
import com.zysd.collection.scheduler.domain.meter.MeterInfoDto;
import com.zysd.collection.scheduler.entity.SchedulerTaskModel;
import com.zysd.collection.scheduler.entity.SchedulerTaskTerminalMapping;
import com.zysd.collection.scheduler.entity.Terminal;
import com.zysd.collection.scheduler.system.enums.CollectTime;
import com.zysd.collection.scheduler.system.util.NamedThreadFactory;
import com.zysd.collection.scheduler.service.amqp.AmqpClientService;
import com.zysd.collection.scheduler.service.model.ModelTerminalMappingService;
import com.zysd.collection.scheduler.service.record.RecordApiService;
import com.zysd.collection.scheduler.system.enums.ExecuteType;
import com.zysd.collection.scheduler.system.enums.Retry;
import com.zysd.collection.scheduler.system.enums.RetryUnit;
import com.zysd.collection.scheduler.system.exception.SchedulerTaskExecuteException;
import org.joda.time.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;

/**
 * 采集任务模板 模板任务执行
 *
 * @author liuht
 * @date 2017/10/12 9:20
 */
@SuppressWarnings("unchecked")
@Component("schedulerTaskForward")
public class SchedulerTaskForward {
    public static final String BEAN_NAME = "schedulerTaskForward";
    public static final String METHOD = "forward";
    private Gson gson = new GsonBuilder().create();
    private static final Logger LOGGER = LoggerFactory.getLogger(SchedulerTaskForward.class);

    @Autowired
    private RecordApiService apiService;

    @Autowired
    private AmqpClientService mqService;

    @Autowired
    private ModelTerminalMappingService modelTerminalMappingService;

    private ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(10, new NamedThreadFactory("collectSchedulerRetry-"));

    @Value("${collectd.batch}")
    protected int batch;

    public void forward(String json) {
        try {
            SchedulerTaskModel model = gson.fromJson(json, SchedulerTaskModel.class);
            if (ExecuteType.report.getCode().equals(Integer.valueOf(model.getExecuteType()))) {
                // 上报的在定时任务中不执行
                throw new SchedulerTaskExecuteException("任务: " + model.getTaskName() + ", 属于上报任务, 禁止在定时任务中执行");
            }
            executeCollect(model.getTaskName(), CollectTime.forValue(model.getCollectTime()), TaskType.forCode(model.getTaskType()), model.getAfns(), model.getId(), model.getTermType());
            // 任务重试验证
            Retry retry = Retry.forCode(Integer.valueOf(model.getMeasureRetryEnable()));
            if (retry != null && retry.ordinal() == Retry.enable.ordinal()) {
                collectRetry(model);
            }
            LOGGER.debug("任务: " + model.getTaskName() + ", 执行完成");
        } catch (Exception e) {
            LOGGER.error("定时任务执行出错!", e);
        }
    }

    /**
     * 采集定时任务重试
     *
     * @param model 采集任务详情
     */
    private void collectRetry(SchedulerTaskModel model) {
        LOGGER.debug("任务: " + model.getTaskName() + ", 需要重试, 正在执行重试逻辑");
        // 重试单位
        RetryUnit retryUnit = RetryUnit.forCode(model.getMeasureRetryUnit());
        // 重试延迟
        int retryDelay = model.getMeasureRetryDelay();
        // 重试间隔
        int retryInterval = model.getMeasureRetryInterval();
        // 重试次数
        int retryTimes = model.getMeasureRetryTimes();
        for (int i = 1; i <= retryTimes; i++) {
            assert retryUnit != null;
            LOGGER.debug("任务: " + model.getTaskName() + ", 需要重试, 重试第[" + i + "]次.");
            scheduledExecutorService.schedule(() -> executeCollect(model.getTaskName(), CollectTime.forValue(model.getCollectTime()), TaskType.forCode(model.getTaskType()), model.getAfns(), model.getId(), model.getTermType()), retryDelay + (retryInterval * i), retryUnit.getTimeUnit());
            if (i == retryTimes) {
                LOGGER.debug("任务: " + model.getTaskName() + ", 重试完成.");
            }
        }
    }

    /**
     * 执行召测任务 终端过滤
     *
     * @param taskName 任务名称
     * @param taskType 任务对象类型
     * @param afns     召测信息点
     * @param modelId  任务模板id
     * @param termType 集中器类型
     */
    private void executeCollect(String taskName, CollectTime collectTime, TaskType taskType, String afns, Long modelId, String termType) {
        // 获取对应的终端
        List<SchedulerTaskTerminalMapping> mappings = modelTerminalMappingService.modelmMappings(modelId);
        if (CollectionUtils.isEmpty(mappings)) {
            LOGGER.warn("任务:{}, 执行失败, 未获取到对应的终端", taskName);
            return;
        }
        for (SchedulerTaskTerminalMapping mapping : mappings) {
            if (StringUtils.isEmpty(mapping.getRegion()) || StringUtils.isEmpty(mapping.getLocal())
                    || StringUtils.isEmpty(mapping.getIpAddr()) || StringUtils.isEmpty(mapping.getIpPort())) {
                LOGGER.warn("终端: {}" + "参数不全, 参数详情region:local:ipAddr:ipPort,{}", mapping.getTerminalName(),
                        mapping.getRegion() + ":" + mapping.getLocal() + ":" + mapping.getIpAddr() + ":" + mapping.getIpPort());
                continue;
            }
            // 暂时不过滤termType
            Terminal terminal = new Terminal(mapping.getRegion(), mapping.getLocal(), mapping.getIpAddr(), Integer.valueOf(mapping.getIpPort()));
            terminal.setId(mapping.getTerminalId());
            executeCollect(taskName, terminal, collectTime,taskType, afns);
        }
    }

    /**
     * 执行召测任务 任务对象类型过滤 获取pns
     *
     * @param taskName 任务名称
     * @param terminal 终端信息
     * @param taskType 任务对象类型
     * @param afns     信息类
     */
    private void executeCollect(String taskName, Terminal terminal, CollectTime collectTime, TaskType taskType, String afns) {
        // 获取终端对应的电能表
        PageResponse<MeterInfoDto> meterPage = apiService.getCusMeters(terminal.getId(), 1, 2048);
        List<MeterInfoDto> meters = meterPage.getRows();
        LOGGER.info("3===================meters.size = " + meters.size());
        if (CollectionUtils.isEmpty(meters)) {
            LOGGER.warn("任务:{}, 执行失败, 未获取到终端对应电能表. 终端: " + terminal.toString(), taskName);
            return;
        }

        List<Integer> pnList = new ArrayList<>(meters.size());
        for (MeterInfoDto meter : meters) {
            pnList.add(meter.getPn());
        }

        // 获取fn 以及对应的afn
        String[] codes = afns.split(",");
        LOGGER.info("4===================afns.size=" + afns.length());
        for (String code : codes) {
            CollectInfos info = CollectInfos.forCode(code);
            Collection<Integer> pns = new ArrayList();
            if (taskType.ordinal() == TaskType.Terminal.ordinal()) {
                // 终端
                pns.add(1);
            } else if (taskType.ordinal() == TaskType.MeasuringPoint.ordinal()) {
                // 测量点
                pns.addAll(pnList);
            }
            executeCollect(terminal, info, pns, collectTime);
            LOGGER.info("任务:{}, afn=" + info.getAfn() + ", pns=" + pns + ", fn=" + info.getInfoFn() +
                    ", region:local=" + terminal.getRegionNo() + ":" + terminal.getLocalNo(), taskName);
        }
    }

    /**
     * 根据CollectInfoGroup 执行下发命令
     *
     * @param terminal 终端信息
     * @param info     信息类
     * @param pns      pn集合
     */
    private void executeCollect(Terminal terminal, CollectInfos info, Collection<Integer> pns, CollectTime collectTime) {
        LocalDate date = null;
        switch (collectTime){
            case currDay:
                date = LocalDate.now();
                break;
            case prevDay:
                date = LocalDate.now().plusDays(-1);
                break;
            default:
                break;
        }

        Pipeline<?> value = DataTypeFactory.newRequest(info.getAfn(), info.getInfoFn());

        switch (info.getGroup()) {
            case day:
                TdDay tdDay = (TdDay) value;
                tdDay.freezeTime = date;
                break;
            case month:
                // 月冻结默认都召上一月
                TdMonth tdMonth = (TdMonth) value;
                tdMonth.freezeTime = YearMonth.now().plusMonths(-1);
                break;
            case realtime:
                break;
            case balance:
                break;
            case curve:
                TdCurve tdCurve = (TdCurve) value;
                tdCurve.freezeTime = LocalDateTime.parse(date.toString());
                tdCurve.thickness = FreezeThickness.Minutes15;
                tdCurve.points = DateTimeConstants.MINUTES_PER_DAY / tdCurve.thickness.minute;
                break;
            default:
                break;
        }

        MetersMultiMessages msgs = new MetersMultiMessages(null, info.getAfn());
        for (int pn : pns) {
            //每batch次发送一起消息
            msgs.messages.add(new Message(info.getInfoFn(), pn, value));
            if (msgs.messages.size() >= batch) {
                mqService.publish(terminal, msgs);
                msgs.messages.clear();
            }
        }
        if (!msgs.messages.isEmpty()) {
            mqService.publish(terminal, msgs);
        }
    }
}
